package services

import (
	"bytes"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"runtime"
	"sync"
	"sync/atomic"
	"time"

	"github.com/sirupsen/logrus"

	"github.com/mia0x75/repeater/config"
)

var tr = &http.Transport{
	MaxIdleConnsPerHost: 64,
	TLSClientConfig: &tls.Config{
		InsecureSkipVerify: true,
	},
	DisableCompression:  false,
	DisableKeepAlives:   false,
	TLSHandshakeTimeout: 10 * time.Second,
	Dial: func(netw, addr string) (net.Conn, error) {
		dial := net.Dialer{
			Timeout:   HTTP_POST_TIMEOUT * time.Second,
			KeepAlive: 86400 * time.Second,
		}
		conn, err := dial.Dial(netw, addr)
		if err != nil {
			return conn, err
		}
		return conn, nil
	},
}

// 超时设置
var defaultHttpClient = http.Client{
	Transport: tr,
}

func Post(addr string, post_data []byte) ([]byte, error) {
	req, err := http.NewRequest("POST", addr, bytes.NewReader(post_data))
	if err != nil {
		return nil, err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Connection", "keep-alive")
	// 执行post
	resp, err := defaultHttpClient.Do(req)
	if err != nil {
		return nil, err
	}
	// 关闭io
	defer resp.Body.Close()
	// 判断返回状态
	if resp.StatusCode != http.StatusOK {
		// 返回异常状态
		return nil, errors.New(fmt.Sprintf("HTTP return code: %d.", resp.StatusCode)) //ErrorHttpStatus
	}
	res, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}
	io.Copy(ioutil.Discard, resp.Body)
	return res, nil
}

// 创建一个新的http服务
func NewHttpService(ctx *context.Context, cfg *config.Config) *HttpService {
	l := len(cfg.Subscribers)
	svc := &HttpService{
		lock:             new(sync.Mutex),
		nodes:            make([]*httpNode, l),
		sendFailureTimes: int64(0),
		enable:           true,
		timeTick:         3, //config.TimeTick, // TODO
		wg:               new(sync.WaitGroup),
		clientsCount:     l,
		ctx:              ctx,
	}
	index := 0
	for _, v := range cfg.Subscribers {
		svc.nodes[index] = &httpNode{
			url:              v.Url,
			sendQueue:        make(chan string, HTTP_MAX_SEND_QUEUE),
			sendTimes:        int64(0),
			sendFailureTimes: int64(0),
			offline:          false,
			lock:             new(sync.Mutex),
			failureTimesFlag: int32(0),
			cacheIsInit:      false,
		}
		index++
	}
	return svc
}

// 开始服务
func (svc *HttpService) Start() {
	if !svc.enable {
		return
	}
	cpu := runtime.NumCPU()
	for _, node := range svc.nodes {
		go svc.errorCheckService(node)
		// 启用cpu数量的服务协程
		for i := 0; i < cpu; i++ {
			svc.wg.Add(1)
			go svc.clientSendService(node)
		}
	}
}

// 初始化节点缓冲区，这个缓冲区用于存放发送失败的数据，最多HTTP_CACHE_LEN条
func (svc *HttpService) cacheInit(node *httpNode) {
	if node.cacheIsInit {
		return
	}
	logrus.Infof("http服务初始化失败重试使用的cache")
	node.cache = make([][]byte, HTTP_CACHE_LEN)
	for k := 0; k < HTTP_CACHE_LEN; k++ {
		node.cache[k] = make([]byte, HTTP_CACHE_BUFFER_SIZE)
	}
	node.cacheIsInit = true
	node.cacheIndex = 0
	node.cacheFull = false
}

// 添加数据到缓冲区
func (svc *HttpService) addCache(node *httpNode, msg []byte) {
	node.cache[node.cacheIndex] = append(node.cache[node.cacheIndex][:0], msg...)
	node.cacheIndex++
	logrus.Debugf("append data to buffer for node: %s.", node.url)
	if node.cacheIndex >= HTTP_CACHE_LEN {
		node.cacheIndex = 0
		node.cacheFull = true
	}
}

// 尝试对失败的数据进行重发
func (svc *HttpService) sendCache(node *httpNode) {
	if node.cacheIndex > 0 {
		//保持时序
		if node.cacheFull {
			for i := node.cacheIndex; i < HTTP_CACHE_LEN; i++ {
				//重发
				logrus.Warn("buffer overflow for node: %s.", node.url)
				node.sendQueue <- string(node.cache[i])
			}

		}
		for i := 0; i < node.cacheIndex; i++ {
			//重发
			logrus.Warnf("data resend for node: %s.", node.url)
			node.sendQueue <- string(node.cache[i])
		}
		node.cacheFull = false
		node.cacheIndex = 0
	}
}

// 节点故障检测与恢复服务
func (svc *HttpService) errorCheckService(node *httpNode) {
	for {
		node.lock.Lock()
		sleepTime := time.Second * svc.timeTick

		if node.offline {
			times := atomic.LoadInt64(&node.errorCheckTimes)
			step := float64(times) / float64(1000)
			if step > float64(1) {
				sleepTime = time.Duration(step) * time.Second
				if sleepTime > 60 {
					sleepTime = 60
				}
			}
			// 发送空包检测
			// post默认3秒超时，所以这里不会死锁
			logrus.Debugf("node %s id down.", node.url)
			_, err := Post(node.url, []byte{byte(0)})
			if err == nil {
				//重新上线
				node.offline = false
				atomic.StoreInt64(&node.errorCheckTimes, 0)
				logrus.Warnf("node %s restored.", node.url)
				//对失败的cache进行重发
				svc.sendCache(node)
			} else {
				logrus.Errorf("node %s failure, error: %+v", node.url, err)
			}
			atomic.AddInt64(&node.errorCheckTimes, 1)
		}
		node.lock.Unlock()
		time.Sleep(sleepTime)
		select {
		case <-(*svc.ctx).Done():
			logrus.Debugf("node health check for %s end.", node.url)
			return
		default:
		}
	}
}

// 节点服务协程
func (svc *HttpService) clientSendService(node *httpNode) {
	defer svc.wg.Done()
	for {
		select {
		case msg, ok := <-node.sendQueue:
			if !ok {
				logrus.Warnf("http服务-发送消息channel通道关闭")
				return
			}
			if !node.offline {
				atomic.AddInt64(&node.sendTimes, int64(1))
				logrus.Debug("sending data %s to node %s", string(msg), node.url)
				data, err := Post(node.url, []byte(msg))
				if err != nil {
					atomic.AddInt64(&svc.sendFailureTimes, int64(1))
					atomic.AddInt64(&node.sendFailureTimes, int64(1))
					atomic.AddInt32(&node.failureTimesFlag, int32(1))
					// 如果连续3次错误，标志位故障
					if atomic.LoadInt32(&node.failureTimesFlag) >= 3 {
						//发生故障
						logrus.Warnf("the failure times for node %s reach the threshold, take it offline.", node.url)
						node.lock.Lock()
						node.offline = true
						node.lock.Unlock()
					}
					logrus.Warnf("node %s is still failure after %d retry(ies).", node.url, node.sendFailureTimes)
					svc.cacheInit(node)
					svc.addCache(node, []byte(msg))
				} else {
					node.lock.Lock()
					if node.offline {
						node.offline = false
					}
					node.lock.Unlock()
					//恢复即时清零故障计数
					if atomic.LoadInt32(&node.failureTimesFlag) > 0 {
						atomic.StoreInt32(&node.failureTimesFlag, 0)
					}
					//对失败的cache进行重发
					svc.sendCache(node)
				}
				logrus.Debug("node %s returns %s.", node.url, string(data))
			} else {
				// 故障节点，缓存需要发送的数据
				// 这里就需要一个map[string][10000][]byte，最多缓存10000条
				// 保持最新的10000条
				svc.addCache(node, []byte(msg))
			}
		case <-(*svc.ctx).Done():
			if len(node.sendQueue) <= 0 {
				logrus.Debugf("data sending for node %s end.", node.url)
				return
			}
		}
	}
}

// 对外的广播发送接口
func (svc *HttpService) SendAll(msg []byte) bool {
	if !svc.enable {
		return false
	}
	logrus.Debug("sending broadcast message: ", string(msg))
	svc.lock.Lock()
	for _, node := range svc.nodes {
		logrus.Debugf("sending data %s to node %s, data: %s", string(msg), node.url)
		if len(node.sendQueue) >= cap(node.sendQueue) {
			logrus.Warnf("buffer for node %s overflow, drop data %s", node.url, string(msg))
			continue
		}
		node.sendQueue <- string(msg)
	}
	svc.lock.Unlock()
	return true
}

func (svc *HttpService) Close() {
	logrus.Debug("http service closing, waiting for buffer send complete.")
	if svc.clientsCount > 0 {
		svc.wg.Wait()
	}
	logrus.Debug("http service closed.")
}
